package com.gj.xxl.job.admin.core.thread;

import com.gj.xxl.job.admin.core.config.GjJobScheduleConfig;
import com.gj.xxl.job.admin.core.entity.JobInfo;
import com.gj.xxl.job.admin.dao.XxlJobInfoDAO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 进行定时任务的调度
 */
public class JobScheduleHelper {

    private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);

    // 单例模式
    private static JobScheduleHelper instance = new JobScheduleHelper();

    public static JobScheduleHelper getInstance() {
        return instance;
    }

    //调度线程
    private Thread scheduleThread;
    //时间轮线程
    private Thread ringThread;

    private volatile boolean scheduleThreadToStop = false;
    private volatile boolean ringThreadToStop = false;

    private boolean existJobToBeExecuted = false;

    // 源码作者写的是 5 s
    private long pre_read_ms = 5 * 1000;

    private ConcurrentHashMap<Integer, List<Integer>> ringMap = new ConcurrentHashMap<>();


    /**
     * 目前只是实现了最基础版本的调度功能，存在非常多的问题
     * todo
     * 1.一直 while 循环查库，太蠢了，而且还是查全量的任务，是不是可以带上时间条件去查询？当然这里具体没有想的这么简单，是很复杂的。
     * 2.执行任务的逻辑也是在调度线程中同步去做的，那么如果这个任务执行耗时比较长呢？那么调度线程就被阻塞住了，那么不久废了吗
     * 所以需要另起一个线程异步的去执行，但是不可能执行一个任务就创建一个线程吧，频繁创建销毁线程的成本是很高的。
     * 那么就引入了线程池！这块还有一个快慢线程池的优化。
     */
    public void start() {
        logger.info(">>>>>>>>>> init scheduleThread success");

        //时间对齐，暂时还没想明白为什么要有这段代码
        //后来问了下 deepseek
        //我们把眼光放长远一点，假如未来调度中心是 HA 高可用部署的呢？
        //为了避免任务被重复调用，只能让一台调度中心机器来进行调度
        //所以还要争抢分布式锁，如果时间都不对齐，那各个调度中心都不是同一时间争抢锁，那分布式锁也就失去作用了
        //但是，上面的理由好像也不是很正确啊

        //我可以理解 ringThread 时间轮线程为什么做时间对齐，是因为刻度是 1 s，每个整秒执行所在刻度下的任务列表。
        //但是为什么 scheduleThread 也需要时间对齐呢，调度线程内部的逻辑不就是不断的查库吗，为什么一定要进行时间对齐，感觉对不对齐，都可以啊
        try {
            Thread.sleep(5000 - System.currentTimeMillis() % 1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        //调度中心集群争抢分布式锁的逻辑很简单，就使用 MySQL 来做的，对应底层的 xxl_job_lock 表
        //select * from xxl_job_lock where lock_name = 'schedule_lock' for update;


        // 调度的逻辑，简单来说，就是查库得到定时任务的信息，然后用当前时间与任务要执行的时间进行比较
        /**
         *  查库的逻辑很值得思考
         *  首先每次都查全量的 job 一定不行，所以要带上时间的范围，以当前时间 now 为基准
         *  那到底是查出 1s 之内的？还是 5s 之内的？还是 10s 之内的？
         *  如果我只查询出 1s 之内要执行的任务，这样每次查出的任务确实很少，而且时间的精度比较高
         *  就比如说有一个任务要在 1s 之后执行，我才在 1s 之前把它从数据库中查出来，然后判断是否应该执行了
         *  这么做看似很好，但是换一种角度思考，频繁的访问数据库，本身对性能就是一种拖累
         *  每次只查询下一秒的定时任务，然后再让线程池去执行，但是假如过，在查询这些任务，访问数据库，或者其他方面出现了阻塞呢？
         *  每次只查询一秒之内的任务，然后判断是否应该执行了，这么一来，后面的任务都会延后执行，会出现任务遗漏的问题
         *  这么一看好像精度也没有这么准确了
         *  如果我查询出 1 分钟之内要执行的任务呢？
         *  这样一来查出来的任务势必就会有很多。
         *  先不说任务调度会不会影响后续任务的远程调用，也不考虑查库耗时可能会很长的问题
         *  就只考虑最实际的问题，如果一次去除 1 分钟之内要执行的任务，当前时间可能才 1 分钟 200 ms，
         *  现在取出了 1 分钟到 2 分钟之间要执行的定时任务，如果有的任务是在 1 分钟 100 ms 要执行的，那很好，直接调度就行了
         *  那如果是 1 分钟 300 ms，1 分钟 50 秒要执行的任务呢？会被判断没到执行时间，然后就继续 while 循环去查，
         *  那么 1 分钟 50 秒的任务，还会不断的被查出来，然后跳过，那把它查出来的意义是什么？
         *  显然，这种决策也是不行的
         *  我们换一种思路，只让 scheduleThread 调度线程负责从数据库中扫描出下一个阶段可以执行的定时任务的信息，
         *  把这些信息存到一个数据结构中，然后让另一个线程到点之后就直接调度它们
         *  这不就是时间轮吗
         */
        scheduleThread = new Thread(() -> {
            while (!scheduleThreadToStop) {
                long start = System.currentTimeMillis();
                try {
                    XxlJobInfoDAO xxlJobInfoDAO = GjJobScheduleConfig.getInstance().getJobInfoDAO();
                    long now = System.currentTimeMillis();
                    List<JobInfo> jobInfoList = xxlJobInfoDAO.preReadJobs(now + pre_read_ms);
                    existJobToBeExecuted = false;
                    for (JobInfo jobInfo : jobInfoList) {
                        existJobToBeExecuted = true;

                        //todo 这里只是为了测试方便，后期需要重构的
                        if (!jobInfo.getScheduleType().equals("FIX_RATE")) {
                            continue;
                        }


                        //分情况讨论，注意讨论细致
                        //假设当前时间是 10s，然后查询出小于 15s 的任务
                        //1.任务的执行时间小于 now 10s，定时任务过期了，misfire 错过了
                        //a.过期时间太久的
                        //b.过期不超过一个调度周期 5s 的
                        //2.任务的执行时间大于 now 10s，这种情况很简单，任务还没有到执行时间，就直接交给时间轮处理即可

                        if (now > jobInfo.getTriggerNextTime() + pre_read_ms) {
                            //直接放弃执行这次了，记得更新下一次的触发时间
                            updateNextTriggerTime(jobInfo, System.currentTimeMillis());
                            xxlJobInfoDAO.updateTriggerTime(jobInfo);
                            logger.info(">>>>>>>>>> job misfire, jobId = " + jobInfo.getId());
                        } else if (now > jobInfo.getTriggerNextTime()) {
                            //直接执行一次
                            JobTriggerHelper.getInstance().trigger(jobInfo);
                            updateNextTriggerTime(jobInfo, System.currentTimeMillis());
                            xxlJobInfoDAO.updateTriggerTime(jobInfo);
                        } else {
                            // 当前时间的秒数，放入对应的槽位上
                            pushJobTimeRing(jobInfo);
                            // 更新下一次的执行时间，可能是 cron 或者固定频率，这里先实现固定频率的
                            updateNextTriggerTime(jobInfo, jobInfo.getTriggerNextTime());
                            xxlJobInfoDAO.updateTriggerTime(jobInfo);
                        }
                    }
                    //scheduleThread 调度线程总不能一直查库吧，需要睡眠控制一下
                    //这里有一个判断，1000毫秒就是1秒，如果总耗时小于1秒，就默认数据库中可能没多少数据
                    //线程就不必工作得那么繁忙，所以下面要让线程休息一会，然后再继续工作
                    long cost = System.currentTimeMillis() - start;
                    logger.info(">>>>>>>>>>>>> scheduleThread 此次调度耗时为: " + cost + "ms");
                    if (cost < 1000) {
                        //如果查到了定时任务，就最多睡眠 1s
                        //如果根本没有查到数据，就最多睡眠 5s
                        Thread.sleep(existJobToBeExecuted ? 1000 : pre_read_ms - System.currentTimeMillis() % 1000);
                    }
                } catch (Exception e) {
                    logger.info(Thread.currentThread().getName() + "线程出现异常了");
                    e.printStackTrace();
                }
            }
            logger.info(">>>>>>>>>> scheduleThread stop");
        });

        scheduleThread.setDaemon(true);
        scheduleThread.setName("JobScheduleThread");
        scheduleThread.start();

        ringThread = new Thread(() -> {
            logger.info(">>>>>>>>>> init ringThread success");
            XxlJobInfoDAO xxlJobInfoDAO = GjJobScheduleConfig.getInstance().getJobInfoDAO();
            while (!ringThreadToStop) {
                //时间轮执行任务是按照时间刻度来执行的，每个整秒开始工作
                try {
                    Thread.sleep(1000 - System.currentTimeMillis() % 1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                try {
                    int sec = (int) (System.currentTimeMillis() / 1000 % 60);
                    List<Integer> jobIdList = ringMap.get(sec);
                    if (jobIdList != null) {
                        for (Integer jobId : jobIdList) {
                            JobInfo jobInfo = xxlJobInfoDAO.findByJobId(jobId);
                            JobTriggerHelper.getInstance().trigger(jobInfo);
                        }
                        jobIdList.clear();
                    }
                } catch (Exception e) {
                    if (!ringThreadToStop) {
                        logger.error("ringThread error");
                    }
                }
            }
            logger.info(">>>>>>>>>> ringThread stop");
        });

        ringThread.setDaemon(true);
        ringThread.setName("ringThread");
        ringThread.start();
    }

    public void destroy() {
        scheduleThreadToStop = true;
        ringThreadToStop = true;
    }

    private void pushJobTimeRing(JobInfo jobInfo) {
        int sec = (int) (jobInfo.getTriggerNextTime() / 1000 % 60);
        List<Integer> jobInfos = ringMap.get(sec);
        if (jobInfos == null) {
            jobInfos = new ArrayList<>();
            ringMap.put(sec, jobInfos);
        }
        jobInfos.add(jobInfo.getId());
    }

    private void updateNextTriggerTime(JobInfo jobInfo, long TriggerTime) {
        jobInfo.setTriggerLastTime(TriggerTime);
        jobInfo.setTriggerNextTime(TriggerTime + Long.parseLong(jobInfo.getScheduleConf()) * 1000);
    }

}
